package com.jxb.sequentialconsumption;

import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

/**
 * 类注释
 *
 * @author jiaobo
 * @date Created in 2024/9/10 21:30
 **/
public class Consumer {

    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer pushConsumer =
                new DefaultMQPushConsumer("please_rename_unique_group_name_3");

        pushConsumer.setNamesrvAddr("127.0.0.1:9876");
        pushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        pushConsumer.subscribe("OrderTopicTest", "*");

        //MessageListenerOrderly：会去锁住队列，一个一个去取
        //MessageListenerConcurrently：不回去锁住队列，每次都会从Messgae取出一批数据，默认不会超过32条
        pushConsumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
                                                       ConsumeOrderlyContext context) {
                context.setAutoCommit(true);
                for(MessageExt msg:msgs){
                    System.out.println("收到消息内容 "+new String(msg.getBody()));
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        pushConsumer.start();
        System.out.printf("Consumer Started.%n");
    }

}
